package com.gitee.sparta.core.process.workers;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.StrUtil;
import com.gitee.sparta.common.SpartaConstant;
import com.gitee.sparta.common.helpers.NetworkDelayHelper;
import com.gitee.sparta.common.helpers.RedisKeyHelper;
import com.gitee.sparta.common.helpers.ScriptShaHelper;
import com.gitee.sparta.core.cluster.balance.LoadBalance;
import com.gitee.sparta.core.process.feature.SpartaTaskDispatcherFeature;
import com.gitee.sparta.core.process.feature.SpartaWorkerFeature;
import com.gitee.sparta.core.process.operate.AbstractTimeoutTabOperate;
import com.gitee.sparta.core.property.SpartaCoreProperties;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RedissonClient;

import java.util.List;

/**
 * sparta 线程开始
 *
 * @author zyred
 * @since v 0.0.1
 */
@Slf4j
public class SpartaCarryWorker<V> extends Thread {
    /**
     * lua 脚本返回结果 list 的索引定义
     * TOPIC_INDEX:  普通队列中执行搬运后的所有 topic，在返回的结果中下标为 0
     * MIN_INDEX：   timeout_tab 中剩余的 topic 中，最小分数
     * CHANNEL_INDEX：   搬运完毕后，是广播的 topic
     */
    private static final int TOPIC_INDEX = 0, MIN_INDEX = 1, CHANNEL_INDEX = 2;
    /** 启动状态 **/
    public static volatile boolean enable = true;

    private final LoadBalance loadBalance;
    private final RedissonClient redissonClient;
    private final SpartaCoreProperties properties;
    private final AbstractTimeoutTabOperate<V> timeoutTabOperate;
    private final SpartaTaskDispatcherFeature<V> notifierFeature;


    public SpartaCarryWorker(LoadBalance loadBalance,
                             RedissonClient redissonClient,
                             SpartaCoreProperties properties,
                             AbstractTimeoutTabOperate<V> timeoutTabOperate,
                             SpartaTaskDispatcherFeature<V> notifierFeature) {
        super.setDaemon(true);
        super.setName(SpartaConstant.SPARTA_READ_THREAD_NAME);
        this.properties = properties;
        this.loadBalance = loadBalance;
        this.redissonClient = redissonClient;
        this.notifierFeature = notifierFeature;
        this.timeoutTabOperate = timeoutTabOperate;
    }

    @Override
    public void run() {

        final String systemName = this.properties.getSystemName();
        final String timeoutTabKey = RedisKeyHelper.tabKey(systemName, SpartaConstant.TABLE_TIMEOUT_NAME);
        final String readableTabKey = RedisKeyHelper.tabKey(systemName, SpartaConstant.TABLE_READABLE_NAME);

        if (log.isDebugEnabled()) {
            log.debug("[Sparta] Carry worker start, system name : {}", systemName);
        }

        while (enable) {
            try {
                int delay = NetworkDelayHelper.read();
                // 工作完毕一次后，拿到 timeout_tab 中剩余的个数里最小的一个分数
                long st = System.currentTimeMillis();
                String balance = this.loadBalance.balance();
                long score = this.doWork(timeoutTabKey, readableTabKey, (st + delay), balance);

                if (delay == 0) {
                    delay = (int) (System.currentTimeMillis() - st);
                    NetworkDelayHelper.write(delay);
                }
                SpartaWorkerFeature.nextWorker(score > 0 ? (score - delay) : score);
            } catch (Exception ex) {
                log.error("[Sparta] Carry, exception : {}", ex.getMessage(), ex);
                ScriptShaHelper.reloadScriptSha(ex, this.redissonClient, this.properties);
                SpartaWorkerFeature.nextWorker((System.currentTimeMillis() + 1000L));
            }
        }
        log.info("[Sparta] Stopped carry thread ...");
    }

    /**
     * 开始处理超时的数据，从 timeout_tab 中将超时的数据转移到 readable_tab 中
     *
     * @param timeoutTabKey  {systemName}:timeout_tab
     * @param readableTabKey {systemName}:readable_tab
     * @return minScore：本次 worker 结束后，timeout_tab 中剩余数据中最小分数
     */
    @SuppressWarnings("all")
    private long doWork(final String timeoutTabKey, final String readableTabKey, final long current, String balance) {
        List<Object> result = this.timeoutTabOperate.carry(timeoutTabKey, readableTabKey, current, balance);

        // 理论上这种情况不会出现
        if (CollUtil.isEmpty(result)) {
            return 0L;
        }

        // 向消费线程延迟队列中投递过期的消息
        String carryTopic = String.valueOf(result.get(TOPIC_INDEX));

        if (StrUtil.isNotBlank(carryTopic)) {
            this.notifierFeature.deliverDelayQueue(carryTopic, balance);
        }
        // 向消费线程延迟队列中投递过期的广播
        String carryChannel = String.valueOf(result.get(CHANNEL_INDEX));
        if (StrUtil.isNotBlank(carryChannel)) {
            this.notifierFeature.deliverBroadcast(carryChannel, balance);
        }

        // 搬运结束后，有值的情况才进行下一个服务的切换
        if (StrUtil.isNotBlank(carryTopic) || StrUtil.isNotBlank(carryChannel)) {
            this.loadBalance.switchover(this.properties.isFocusCarry(), 1);
        }
        return Long.valueOf(String.valueOf(result.get(MIN_INDEX)));
    }


    /**
     * 如果线程处于睡眠状态，立刻结束睡眠
     * 如果线程处于工作状态，等线程执行完毕后下次搬运就结束
     */
    public void shutdown() {
        SpartaWorkerFeature.interruptWorker();
        enable = false;
    }
}
